package com.sand.pro.flinksql_pro.work;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public final class OrderPayDetailApp {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3); // source only supports parallelism of 1
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, EnvironmentSettings.newInstance()
                // Watermark is only supported in blink planner
                .useBlinkPlanner()
                .inStreamingMode()
                .build());
        env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
        // access flink configuration
        Configuration configuration = tEnv.getConfig().getConfiguration();
        configuration.setString("table.exec.source.idle-timeout" , "1000");
        // register a table in the catalog
        //latest-offset  earliest-offset
        tEnv.executeSql("CREATE TABLE sdb_order_info (\n" +
                "  id int,\n" +
                "  order_serial string,\n" +
                "  access_mid string,\n" +
                "  customer_order_no string,\n" +
                "  ori_customer_order_no string,\n" +
                "  order_amt DECIMAL(24,2),\n" +
                "  `method` string,\n" +
                "  order_title string,\n" +
                "  order_desc string,\n" +
                "  front_url string,\n" +
                "  back_url string,\n" +
                "  currency string,\n" +
                "  user_no string,\n" +
                "  channel_no string,\n" +
                "  receive_user_no string,\n" +
                "  access_system_no string ,\n" +
                "  order_begin string,\n" +
                "  order_end string ,\n" +
                "  order_time_out string,\n" +
                "  trans_kind string,\n" +
                "  trans_type string,\n" +
                "  buss_type string ,\n" +
                "  trade_status string,\n" +
                "  refund_amt DECIMAL(24,2),\n" +
                "  resp_code string,\n" +
                "  resp_msg string,\n" +
                "  sand_tn string,\n" +
                "  trace_id string,\n" +
                "  market_product_no string,\n" +
                "  work_date string ,\n" +
                "  receive_card_no string,\n" +
                "  memo string,\n" +
                "  func_product_no string,\n" +
                "  remark string,\n" +
                "  merch_order_no string,\n" +
                "  orig_merch_order_no string,\n" +
                "  instg_id string,\n" +
                "  trx_id string,\n" +
                "  store_id string,\n" +
                "  terminal_id string,\n" +
                "  front_trans_code string ,\n" +
                "  merch_goods string\n" +
                ") with("+ MyKafkaUtil.getKafkaDDL("maxwell87_psdb_sdb_order_info","orderinfo_groupid")+")");


        tEnv.executeSql("CREATE TABLE sdb_pay_info (\n" +
                "  id int,\n" +
                "  pay_serial string,\n" +
                "  order_serial string,\n" +
                "  access_mid string,\n" +
                "  customer_order_no string,\n" +
                "  repeat_flag string,\n" +
                "  status string,\n" +
                "  pay_begin_time string,\n" +
                "  pay_end_time string,\n" +
                "  pay_tool string,\n" +
                "  bank_code string,\n" +
                "  credit_flag string,\n" +
                "  order_amt DECIMAL(24,2),\n" +
                "  pay_amt DECIMAL(24,2),\n" +
                "  rebate_amt DECIMAL(24,2),\n" +
                "  fee_amt DECIMAL(24,2),\n" +
                "  channel_resp_code string,\n" +
                "  channel_resp_msg string,\n" +
                "  bindid string,\n" +
                "  pay_card_no string,\n" +
                "  merch_order_no string,\n" +
                "  pay_channel string,\n" +
                "  pay_detail string,\n" +
                "  market_amt DECIMAL(24,2),\n" +
                "  coupon_code string,\n" +
                "  activiy_no string\n" +
                ") with("+ MyKafkaUtil.getKafkaDDL("maxwell87_psdb_sdb_pay_info","payinfo_groupid")+")");

        tEnv.executeSql("CREATE TABLE issue_cust_change_func (\n" +
                "  acc_change_serial_no string,\n" +
                "  access_sys_no string,\n" +
                "  access_serial_no string,\n" +
                "  trans_serial_no string,\n" +
                "  issue_serial_no string,\n" +
                "  customer_order_no string,\n" +
                "  orig_customer_order_no string,\n" +
                "  detail_serial_no string,\n" +
                "  acc_voucher string,\n" +
                "  customer_no string,\n" +
                "  user_no string,\n" +
                "  trans_code string,\n" +
                "  acc_product_no string,\n" +
                "  busi_no string,\n" +
                "  account_org string,\n" +
                "  busi_platform string,\n" +
                "  account_no string,\n" +
                "  bal_direction string,\n" +
                "  acc_direction string,\n" +
                "  acc_amt DECIMAL(14,2),\n" +
                "  after_acc_bal DECIMAL(14,2),\n" +
                "  sys_acc_date string,\n" +
                "  acc_abstract string,\n" +
                "  is_user_show string,\n" +
                "  remark string,\n" +
                "  tamper string,\n" +
                "  insert_time string,\n" +
                "  acas_acc_serial_no string,\n" +
                "  busi_serial_no string,\n" +
                "  acc_role_no string,\n" +
                "  acc_order_no string,\n" +
                "  acc_payer string,\n" +
                "  fund_type string,\n" +
                "  acc_coeff string,\n" +
                "  sys_acc_time string,\n" +
                "  post_script string,\n" +
                "  trans_rival_type string,\n" +
                "  trans_rival_account string,\n" +
                "  trans_rival_account_name string,\n" +
                "  trans_rival_desc string,\n" +
                "  acas_trans_no string,\n" +
                "  acas_busi_no string\n" +
                ") with("+ MyKafkaUtil.getKafkaDDL("ISSUE_CUST_CHANGE_FUNC","issue_change_groupid")+")");


        tEnv.executeSql("CREATE TABLE issue_cust_change_settle (\n" +
                "  acc_change_serial_no string,\n" +
                "  access_sys_no string,\n" +
                "  access_serial_no string,\n" +
                "  trans_serial_no string,\n" +
                "  issue_serial_no string,\n" +
                "  customer_order_no string,\n" +
                "  orig_customer_order_no string,\n" +
                "  detail_serial_no string,\n" +
                "  acc_voucher string,\n" +
                "  customer_no string,\n" +
                "  user_no string,\n" +
                "  trans_code string,\n" +
                "  acc_product_no string,\n" +
                "  busi_no string,\n" +
                "  account_org string,\n" +
                "  busi_platform string,\n" +
                "  account_no string,\n" +
                "  bal_direction string,\n" +
                "  acc_direction string,\n" +
                "  acc_amt DECIMAL(14,2),\n" +
                "  after_acc_bal DECIMAL(14,2),\n" +
                "  sys_acc_date string,\n" +
                "  acc_abstract string,\n" +
                "  is_user_show string,\n" +
                "  remark string,\n" +
                "  tamper string,\n" +
                "  insert_time string,\n" +
                "  acas_acc_serial_no string,\n" +
                "  busi_serial_no string,\n" +
                "  acc_role_no string,\n" +
                "  acc_order_no string,\n" +
                "  acc_payer string,\n" +
                "  fund_type string,\n" +
                "  acc_coeff string,\n" +
                "  sys_acc_time string,\n" +
                "  post_script string,\n" +
                "  trans_rival_type string,\n" +
                "  trans_rival_account string,\n" +
                "  trans_rival_account_name string,\n" +
                "  trans_rival_desc string,\n" +
                "  acas_trans_no string,\n" +
                "  acas_busi_no string\n" +
                ") with("+ MyKafkaUtil.getKafkaDDL("ISSUE_CUST_CHANGE_SETTLE","issue_settle_groupid")+")");

        tEnv.executeSql("CREATE TABLE ums_info_user_base (\n" +
                "  user_id string,\n" +
                "  user_inner_id string,\n" +
                "  device_id string,\n" +
                "  customer_id string,\n" +
                "  user_name string,\n" +
                "  phone_no string,\n" +
                "  email string,\n" +
                "  nick_name string,\n" +
                "  login_pass string,\n" +
                "  current_day_error_time DECIMAL(38,2),\n" +
                "  all_error_time DECIMAL(38,2),\n" +
                "  lock_flag string,\n" +
                "  phone_login_flag string,\n" +
                "  email_login_flag string,\n" +
                "  user_stat string,\n" +
                "  real_name_flag string,\n" +
                "  safe_question_flag string,\n" +
                "  reg_time string,\n" +
                "  welcome_msg string,\n" +
                "  image_url string,\n" +
                "  insert_time string,\n" +
                "  update_time string,\n" +
                "  update_user string,\n" +
                "  pay_for_another_flag string,\n" +
                "  pay_pass_flag string,\n" +
                "  top_id string,\n" +
                "  lock_time string,\n" +
                "  frozen_time string,\n" +
                "  bind_card_flag string,\n" +
                "  small_exemption string,\n" +
                "  personal_rating string,\n" +
                "  live_face_flag string,\n" +
                "  byf_flag string,\n" +
                "  risk_live_face_flag string,\n" +
                "  work string,\n" +
                "  gestures_password string,\n" +
                "  gestures_flag string\n" +
                ") with("+ MyKafkaUtil.getKafkaDDL("UMS_INFO_USER_BASE","user_base_groupid")+")");


        tEnv.executeSql("CREATE TABLE acas_trs_order_jnl (\n" +
                "  local_atoj_serial_no string,\n" +
                "  local_ataj_serial_no string,\n" +
                "  busi_serial_no string,\n" +
                "  access_system_no string,\n" +
                "  access_serial_no string,\n" +
                "  launch_system_no string,\n" +
                "  work_date string,\n" +
                "  batch_no string,\n" +
                "  customer_order_no string,\n" +
                "  orig_customer_order_no string,\n" +
                "  customer_no string,\n" +
                "  user_no string,\n" +
                "  acc_product_no string,\n" +
                "  busi_no string,\n" +
                "  trans_no string,\n" +
                "  customer_trans_name string,\n" +
                "  acc_trans_no string,\n" +
                "  acc_trans_sub_no string,\n" +
                "  account_no string,\n" +
                "  received_user_no string,\n" +
                "  received_acc_product_no string,\n" +
                "  received_account_no string,\n" +
                "  response_code string,\n" +
                "  response_desc string,\n" +
                "  response_status string,\n" +
                "  order_status string,\n" +
                "  order_title string,\n" +
                "  order_create_time string,\n" +
                "  order_time_out string,\n" +
                "  currency string,\n" +
                "  order_amt DECIMAL(14,2),\n" +
                "  trans_amt DECIMAL(14,2),\n" +
                "  rebate_amt DECIMAL(14,2),\n" +
                "  deal_amt DECIMAL(14,2),\n" +
                "  acc_amt DECIMAL(14,2),\n" +
                "  refund_amt DECIMAL(14,2),\n" +
                "  special_cycle_flag string,\n" +
                "  fee_calculate_model string,\n" +
                "  user_fee_flag string,\n" +
                "  user_fee_info string,\n" +
                "  test_flag string,\n" +
                "  synchronise_flag string,\n" +
                "  busi_platform string,\n" +
                "  account_org string,\n" +
                "  order_desc string,\n" +
                "  shop_order_no string,\n" +
                "  shop_order_title string,\n" +
                "  trans_rival_type string,\n" +
                "  trans_rival_account string,\n" +
                "  trans_rival_account_name string,\n" +
                "  channel_refer_no string,\n" +
                "  status string,\n" +
                "  remark string,\n" +
                "  create_time string,\n" +
                "  update_time string,\n" +
                "  resv_amt1 string,\n" +
                "  resv_amt2 string,\n" +
                "  resv1 string,\n" +
                "  resv2 string,\n" +
                "  acc_role_no string,\n" +
                "  received_acc_role_no string,\n" +
                "  postscript string,\n" +
                "  trans_abstract string,\n" +
                "  trans_request_time string,\n" +
                "  trans_rival_desc string,\n" +
                "  channel_no string,\n" +
                "  mktg_activity_info string,\n" +
                "  acas_trans_no string,\n" +
                "  acas_busi_no string,\n" +
                "  trustee_master_account string,\n" +
                "  `extend` string\n" +
                ") with("+ MyKafkaUtil.getKafkaDDL("ACAS_TRS_ORDER_JNL","trs_order_groupid")+")");


        tEnv.executeSql("CREATE TABLE issue_comm_change_mktg (\n" +
                "  acc_change_serial_no string,\n" +
                "  access_sys_no string,\n" +
                "  access_serial_no string,\n" +
                "  acas_acc_serial_no string,\n" +
                "  trans_serial_no string,\n" +
                "  issue_serial_no string,\n" +
                "  customer_order_no string,\n" +
                "  orig_customer_order_no string,\n" +
                "  detail_serial_no string,\n" +
                "  acc_voucher string,\n" +
                "  customer_no string,\n" +
                "  user_no string,\n" +
                "  trans_code string,\n" +
                "  acc_product_no string,\n" +
                "  acc_role_no string,\n" +
                "  mktg_activity_no string,\n" +
                "  busi_no string,\n" +
                "  account_org string,\n" +
                "  busi_platform string,\n" +
                "  account_no string,\n" +
                "  bal_direction string,\n" +
                "  acc_direction string,\n" +
                "  acc_amt DECIMAL(14,2),\n" +
                "  after_acc_bal DECIMAL(14,2),\n" +
                "  sys_acc_date string,\n" +
                "  acc_abstract string,\n" +
                "  is_user_show string,\n" +
                "  remark string,\n" +
                "  tamper string,\n" +
                "  insert_time string,\n" +
                "  trans_rival_type string,\n" +
                "  trans_rival_account string,\n" +
                "  trans_rival_account_name string,\n" +
                "  trans_rival_desc string,\n" +
                "  busi_serial_no string,\n" +
                "  acc_order_no string,\n" +
                "  acc_payer string,\n" +
                "  fund_type string,\n" +
                "  acc_coeff string,\n" +
                "  sys_acc_time string,\n" +
                "  post_script string,\n" +
                "  acas_trans_no string,\n" +
                "  acas_busi_no string\n" +
                ") with("+ MyKafkaUtil.getKafkaDDL("ISSUE_COMM_CHANGE_MKTG","change_mktg_groupid")+")");


        tEnv.executeSql("CREATE VIEW oj AS\n" +
                " SELECT DISTINCT sdb_order_info.id,\n" +
                "            sdb_order_info.order_serial,\n" +
                "            sdb_order_info.access_mid,\n" +
                "            sdb_order_info.customer_order_no,\n" +
                "            sdb_order_info.ori_customer_order_no,\n" +
                "            sdb_order_info.orig_merch_order_no,\n" +
                "            sdb_order_info.order_amt,\n" +
                "            sdb_order_info.`method`,\n" +
                "            sdb_order_info.order_title,\n" +
                "            sdb_order_info.order_desc,\n" +
                "            sdb_order_info.front_url,\n" +
                "            sdb_order_info.back_url,\n" +
                "            sdb_order_info.currency,\n" +
                "            sdb_order_info.user_no,\n" +
                "            sdb_order_info.channel_no,\n" +
                "            sdb_order_info.receive_user_no,\n" +
                "            sdb_order_info.access_system_no,\n" +
                "            sdb_order_info.order_begin,\n" +
                "            sdb_order_info.order_end,\n" +
                "            sdb_order_info.order_time_out,\n" +
                "            sdb_order_info.trans_kind,\n" +
                "            sdb_order_info.trans_type,\n" +
                "            sdb_order_info.front_trans_code,\n" +
                "            sdb_order_info.buss_type,\n" +
                "            sdb_order_info.trade_status,\n" +
                "            sdb_order_info.refund_amt,\n" +
                "            sdb_order_info.resp_code,\n" +
                "            sdb_order_info.resp_msg,\n" +
                "            sdb_order_info.sand_tn,\n" +
                "            sdb_order_info.trace_id,\n" +
                "            sdb_order_info.market_product_no,\n" +
                "            sdb_order_info.work_date,\n" +
                "            sdb_order_info.receive_card_no,\n" +
                "            sdb_order_info.memo,\n" +
                "            sdb_order_info.func_product_no,\n" +
                "            sdb_order_info.merch_order_no\n" +
                "           FROM sdb_order_info\n" +
                "          WHERE ((NOT (EXISTS ( SELECT 1\n" +
                "                   FROM sdb_pay_info\n" +
                "                  WHERE (((sdb_pay_info.pay_tool like '%,%')\n" +
                "                  OR ((sdb_pay_info.pay_tool = '1016')\n" +
                "                  AND (sdb_order_info.trans_kind = '0001')))\n" +
                "                  AND ((sdb_order_info.order_serial) = (sdb_pay_info.order_serial))))))\n" +
                "                  )");


        tEnv.executeSql("CREATE VIEW iss AS\n" +
                "   SELECT substr((issue_cust_change_func.account_no), 24, 1)AS acc_type,\n" +
                "            issue_cust_change_func.acc_change_serial_no,\n" +
                "            issue_cust_change_func.access_sys_no,\n" +
                "            issue_cust_change_func.access_serial_no,\n" +
                "            issue_cust_change_func.trans_serial_no,\n" +
                "            issue_cust_change_func.issue_serial_no,\n" +
                "            issue_cust_change_func.customer_order_no,\n" +
                "            issue_cust_change_func.orig_customer_order_no,\n" +
                "            issue_cust_change_func.detail_serial_no,\n" +
                "            issue_cust_change_func.acc_voucher,\n" +
                "            issue_cust_change_func.customer_no,\n" +
                "            issue_cust_change_func.user_no,\n" +
                "            issue_cust_change_func.trans_code,\n" +
                "            issue_cust_change_func.acc_product_no,\n" +
                "            issue_cust_change_func.busi_no,\n" +
                "            issue_cust_change_func.account_org,\n" +
                "            issue_cust_change_func.busi_platform,\n" +
                "            issue_cust_change_func.account_no,\n" +
                "            issue_cust_change_func.bal_direction,\n" +
                "            issue_cust_change_func.acc_direction,\n" +
                "            issue_cust_change_func.acc_amt,\n" +
                "            issue_cust_change_func.after_acc_bal,\n" +
                "            issue_cust_change_func.sys_acc_date,\n" +
                "            issue_cust_change_func.acc_abstract,\n" +
                "            issue_cust_change_func.is_user_show,\n" +
                "            issue_cust_change_func.remark,\n" +
                "            issue_cust_change_func.tamper,\n" +
                "            issue_cust_change_func.insert_time,\n" +
                "            issue_cust_change_func.busi_serial_no,\n" +
                "            issue_cust_change_func.acc_role_no,\n" +
                "            issue_cust_change_func.acc_order_no,\n" +
                "            issue_cust_change_func.acc_payer,\n" +
                "            issue_cust_change_func.fund_type,\n" +
                "            issue_cust_change_func.acc_coeff,\n" +
                "            issue_cust_change_func.sys_acc_time,\n" +
                "            issue_cust_change_func.post_script,\n" +
                "            issue_cust_change_func.acas_acc_serial_no,\n" +
                "            issue_cust_change_func.trans_rival_type,\n" +
                "            issue_cust_change_func.trans_rival_account,\n" +
                "            issue_cust_change_func.trans_rival_account_name,\n" +
                "            issue_cust_change_func.trans_rival_desc\n" +
                "           FROM issue_cust_change_func\n" +
                "        UNION ALL\n" +
                "         SELECT '' AS acc_type,\n" +
                "            issue_cust_change_settle.acc_change_serial_no,\n" +
                "            issue_cust_change_settle.access_sys_no,\n" +
                "            issue_cust_change_settle.access_serial_no,\n" +
                "            issue_cust_change_settle.trans_serial_no,\n" +
                "            issue_cust_change_settle.issue_serial_no,\n" +
                "            issue_cust_change_settle.customer_order_no,\n" +
                "            issue_cust_change_settle.orig_customer_order_no,\n" +
                "            issue_cust_change_settle.detail_serial_no,\n" +
                "            issue_cust_change_settle.acc_voucher,\n" +
                "            issue_cust_change_settle.customer_no,\n" +
                "            issue_cust_change_settle.user_no,\n" +
                "            issue_cust_change_settle.trans_code,\n" +
                "            issue_cust_change_settle.acc_product_no,\n" +
                "            issue_cust_change_settle.busi_no,\n" +
                "            issue_cust_change_settle.account_org,\n" +
                "            issue_cust_change_settle.busi_platform,\n" +
                "            issue_cust_change_settle.account_no,\n" +
                "            issue_cust_change_settle.bal_direction,\n" +
                "            issue_cust_change_settle.acc_direction,\n" +
                "            issue_cust_change_settle.acc_amt,\n" +
                "            issue_cust_change_settle.after_acc_bal,\n" +
                "            issue_cust_change_settle.sys_acc_date,\n" +
                "            issue_cust_change_settle.acc_abstract,\n" +
                "            issue_cust_change_settle.is_user_show,\n" +
                "            issue_cust_change_settle.remark,\n" +
                "            issue_cust_change_settle.tamper,\n" +
                "            issue_cust_change_settle.insert_time,\n" +
                "            issue_cust_change_settle.busi_serial_no,\n" +
                "            issue_cust_change_settle.acc_role_no,\n" +
                "            issue_cust_change_settle.acc_order_no,\n" +
                "            issue_cust_change_settle.acc_payer,\n" +
                "            issue_cust_change_settle.fund_type,\n" +
                "            issue_cust_change_settle.acc_coeff,\n" +
                "            issue_cust_change_settle.sys_acc_time,\n" +
                "            issue_cust_change_settle.post_script,\n" +
                "            issue_cust_change_settle.acas_acc_serial_no,\n" +
                "            issue_cust_change_settle.trans_rival_type,\n" +
                "            issue_cust_change_settle.trans_rival_account,\n" +
                "            issue_cust_change_settle.trans_rival_account_name,\n" +
                "            issue_cust_change_settle.trans_rival_desc\n" +
                "           FROM issue_cust_change_settle");

        // define a dynamic aggregating query
        tEnv.executeSql("create  view v_pssdb_reconli as  SELECT DISTINCT oj.sand_tn,\n" +
                "    iss.acc_type,\n" +
                "    oj.memo,\n" +
                "    oj.currency,\n" +
                "    oj.order_title,\n" +
                "    oj.order_time_out,\n" +
                "    '' AS merch_name,\n" +
                "    oj.order_end,\n" +
                "        CASE WHEN (oj.trans_kind = '0001') THEN ((0) - oj.order_amt)\n" +
                "        ELSE oj.order_amt\n" +
                "        END AS order_amt,\n" +
                "    oj.trans_type AS trans_type_in,\n" +
                "    oj.trans_type AS trans_type_out,\n" +
                "    oj.order_begin,\n" +
                "    oj.order_serial,\n" +
                "    oj.order_desc,\n" +
                "   '' as  trade_status,\n" +
                "    oj.resp_code,\n" +
                "    oj.resp_msg,\n" +
                "    oj.trans_kind,\n" +
                "    oj.trans_type,\n" +
                "    oj.front_trans_code,\n" +
                "    oj.access_system_no,\n" +
                "    ( SELECT pj_1.credit_flag\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS credit_flag,\n" +
                "    ( SELECT pj_1.bindid\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS bindid,\n" +
                "    ( SELECT pj_1.fee_amt\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS fee_amt,\n" +
                "    ( SELECT pj_1.status\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS status,\n" +
                "    ( SELECT pj_1.channel_resp_code\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS channel_resp_code,\n" +
                "    ( SELECT pj_1.channel_resp_msg\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS channel_resp_msg,\n" +
                "    ( SELECT pj_1.pay_amt\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_amt,\n" +
                "    0.0 AS rebate_amt,\n" +
                "    ( SELECT pj_1.bank_code\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS bank_code,\n" +
                "    ( SELECT pj_1.merch_order_no AS channel_order_no\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS channel_order_no,\n" +
                "    ( SELECT pj_1.pay_begin_time\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_begin_time,\n" +
                "    '' AS pay_tool,\n" +
                "    ( SELECT pj_1.pay_end_time\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_end_time,\n" +
                "    ( SELECT pj_1.pay_card_no\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_card_no,\n" +
                "    ( SELECT pj_1.pay_serial\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_serial,\n" +
                "    ( SELECT pj_1.pay_channel\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_channel,\n" +
                "    ( SELECT pj_1.pay_detail\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_detail,\n" +
                "    oj.customer_order_no,\n" +
                "    oj.access_mid,\n" +
                "    oj.buss_type,\n" +
                "    oj.market_product_no,\n" +
                "    oj.receive_user_no,\n" +
                "        CASE\n" +
                "            WHEN (oj.trans_kind = '0006') THEN oj.receive_user_no\n" +
                "            WHEN (oj.trans_type = '0014') THEN oj.receive_user_no\n" +
                "            ELSE oj.user_no\n" +
                "        END AS user_no,\n" +
                "        CASE\n" +
                "            WHEN (oj.trans_kind = '0006') THEN usr.user_name\n" +
                "            WHEN (oj.trans_type = '0014') THEN usr.user_name\n" +
                "            ELSE us.user_name\n" +
                "        END AS user_name,\n" +
                "    '' AS receive_user_name,\n" +
                "    iss.trans_rival_desc AS trans_rival_type,\n" +
                "     '' as  trans_name,\n" +
                "        CASE\n" +
                "            WHEN (((oj.trans_kind = '0002') OR (oj.trans_kind = '0004')) AND (atoj.order_status = '00') AND (CHAR_LENGTH((iss.trans_rival_account)) > 1)) THEN (concat_ws('', iss.trans_rival_desc, '(', substring((iss.trans_rival_account), (CHAR_LENGTH((iss.trans_rival_account)) - 3), 4), ')'))\n" +
                "            WHEN ((oj.trans_kind = '0003') AND (atoj.order_status = '00') AND (CHAR_LENGTH((iss.trans_rival_account)) = 13) AND (substring((iss.trans_rival_account), 1, 1) = '1')) THEN (concat_ws('', iss.trans_rival_account_name, '(', substring((iss.trans_rival_account), 2, 3), '****', substring((iss.trans_rival_account), 9, 4), ')'))\n" +
                "            WHEN ((oj.trans_kind = '0003') AND (atoj.order_status = '00') AND (CHAR_LENGTH((iss.trans_rival_account)) = 13) AND (substring((iss.trans_rival_account), 1, 1) <> '1')) THEN (concat_ws('', iss.trans_rival_account_name, '(', substring((iss.trans_rival_account), 1, 4), '****', substring((iss.trans_rival_account), 10, 4), ')'))\n" +
                "            WHEN ((oj.trans_type = '0017') AND (atoj.order_status = '00') AND (CHAR_LENGTH((iss.trans_rival_account)) = 13)) THEN (concat_ws('', iss.trans_rival_account_name, '(', substring((iss.trans_rival_account), 2, 3), '****', substring((iss.trans_rival_account), 9, 4), ')'))\n" +
                "            WHEN ((oj.trans_type = '0030') AND (atoj.order_status = '00') AND (CHAR_LENGTH((iss.trans_rival_account)) = 13)) THEN (concat_ws('', iss.trans_rival_account_name, '(', substring((iss.trans_rival_account), 2, 3), '****', substring((iss.trans_rival_account), 9, 4), ')'))\n" +
                "            ELSE iss.trans_rival_desc\n" +
                "        END  as  trans_rival_type_account,\n" +
                "    iss.acc_abstract,\n" +
                "    iss.acc_product_no,\n" +
                "    oj.merch_order_no,\n" +
                "    '' AS create_time,\n" +
                "        CASE\n" +
                "            WHEN ((concat_ws('', iss.acc_direction, iss.bal_direction) in ('DD', 'CC')) AND (oj.trans_kind not in ('0002', '0004'))) THEN iss.acc_amt\n" +
                "            WHEN ((concat_ws('', iss.acc_direction, iss.bal_direction) in ('CD', 'DC')) AND (oj.trans_kind not in ('0002', '0004'))) THEN 0-iss.acc_amt\n" +
                "            WHEN (oj.trans_kind = '0001') THEN 0-oj.order_amt\n" +
                "            WHEN (oj.trans_kind = '0006') THEN oj.order_amt\n" +
                "            ELSE (oj.order_amt)\n" +
                "        END AS acc_amt,\n" +
                "    COALESCE(atoj.postscript, atoj.customer_trans_name) AS postscript,\n" +
                "        CASE oj.trans_kind\n" +
                "            WHEN '0006' THEN oj.orig_merch_order_no\n" +
                "            ELSE oj.ori_customer_order_no\n" +
                "        END AS ori_customer_order_no,\n" +
                "    oj.receive_card_no\n" +
                "   FROM oj \n" +
                "     LEFT JOIN iss ON (((((( iss.customer_order_no ) = ( oj.customer_order_no )) AND ( oj.trans_kind <> '0001' )) OR ((( oj.merch_order_no ) = ( iss.customer_order_no )) \n" +
                " AND ( oj.trans_kind = '0001' ))) \n" +
                " AND (\n" +
                " concat_ws( '', iss.trans_code, iss.bal_direction, iss.acc_direction ) NOT IN ( '3320CC', '3130CC', '3320DD' )) \n" +
                "AND ( iss.fund_type <> '1006' )))"+
                "     LEFT JOIN ums_info_user_base us ON oj.user_no = us.user_id\n" +
                "     LEFT JOIN ums_info_user_base usr ON oj.receive_user_no = usr.user_id\n" +
                "     LEFT JOIN acas_trs_order_jnl atoj ON iss.acas_acc_serial_no = atoj.local_atoj_serial_no AND atoj.acc_trans_sub_no <> '0201'\n"+
                "     ");



        //=====================================================================================
        tEnv.executeSql("create  view oj2 as " +
                "SELECT DISTINCT sdb_order_info.id,\n" +
                "            sdb_order_info.order_serial,\n" +
                "            sdb_order_info.access_mid,\n" +
                "            sdb_order_info.customer_order_no,\n" +
                "            sdb_order_info.ori_customer_order_no,\n" +
                "            sdb_order_info.orig_merch_order_no,\n" +
                "            sdb_order_info.order_amt,\n" +
                "            sdb_order_info.`method`,\n" +
                "            sdb_order_info.order_title,\n" +
                "            sdb_order_info.order_desc,\n" +
                "            sdb_order_info.front_url,\n" +
                "            sdb_order_info.back_url,\n" +
                "            sdb_order_info.currency,\n" +
                "            sdb_order_info.user_no,\n" +
                "            sdb_order_info.channel_no,\n" +
                "            sdb_order_info.receive_user_no,\n" +
                "            sdb_order_info.access_system_no,\n" +
                "            sdb_order_info.order_begin,\n" +
                "            sdb_order_info.order_end,\n" +
                "            sdb_order_info.order_time_out,\n" +
                "            sdb_order_info.trans_kind,\n" +
                "            sdb_order_info.trans_type,\n" +
                "            sdb_order_info.front_trans_code,\n" +
                "            sdb_order_info.buss_type,\n" +
                "            sdb_order_info.trade_status,\n" +
                "            sdb_order_info.refund_amt,\n" +
                "            sdb_order_info.resp_code,\n" +
                "            sdb_order_info.resp_msg,\n" +
                "            sdb_order_info.sand_tn,\n" +
                "            sdb_order_info.trace_id,\n" +
                "            sdb_order_info.market_product_no,\n" +
                "            sdb_order_info.work_date,\n" +
                "            sdb_order_info.receive_card_no,\n" +
                "            sdb_order_info.memo,\n" +
                "            sdb_order_info.func_product_no,\n" +
                "            sdb_order_info.merch_order_no\n" +
                "           FROM sdb_order_info\n" +
                "          WHERE ((EXISTS ( SELECT 1\n" +
                "                   FROM sdb_pay_info\n" +
                "                  WHERE ((((sdb_pay_info.pay_tool) like '%,%') OR (((sdb_pay_info.pay_tool) = '1016') AND (sdb_order_info.trans_kind = '0001'))) AND ((sdb_order_info.order_serial) = (sdb_pay_info.order_serial))))))");
        tEnv.executeSql("create  view v_pssdb_reconli2 as " +
                "SELECT DISTINCT oj.sand_tn,\n" +
                "    '' AS acc_type,\n" +
                "    oj.memo,\n" +
                "    oj.currency,\n" +
                "    oj.order_title,\n" +
                "    oj.order_time_out,\n" +
                "    '' AS merch_name,\n" +
                "    oj.order_end,\n" +
                "    ((0) - oj.order_amt) AS order_amt,\n" +
                "    oj.trans_type AS trans_type_in,\n" +
                "    oj.trans_type AS trans_type_out,\n" +
                "    oj.order_begin,\n" +
                "    oj.order_serial,\n" +
                "    oj.order_desc,\n" +
                "    oj.trade_status,\n" +
                "    oj.resp_code,\n" +
                "    oj.resp_msg,\n" +
                "    oj.trans_kind,\n" +
                "    oj.trans_type,\n" +
                "    oj.front_trans_code,\n" +
                "    oj.access_system_no,\n" +
                "    ( SELECT pj_1.credit_flag\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS credit_flag,\n" +
                "    ( SELECT pj_1.bindid\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS bindid,\n" +
                "    ( SELECT pj_1.fee_amt\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS fee_amt,\n" +
                "    ( SELECT pj_1.status\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS status,\n" +
                "    ( SELECT pj_1.channel_resp_code\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS channel_resp_code,\n" +
                "    ( SELECT pj_1.channel_resp_msg\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS channel_resp_msg,\n" +
                "    ( SELECT pj_1.pay_amt\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_amt,\n" +
                "    0.00 AS rebate_amt,\n" +
                "    ( SELECT pj_1.bank_code\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS bank_code,\n" +
                "    ( SELECT pj_1.merch_order_no AS channel_order_no\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS channel_order_no,\n" +
                "    ( SELECT pj_1.pay_begin_time\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_begin_time,\n" +
                "    '' AS pay_tool,\n" +
                "    ( SELECT pj_1.pay_end_time\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_end_time,\n" +
                "    ( SELECT pj_1.pay_card_no\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_card_no,\n" +
                "    ( SELECT pj_1.pay_serial\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_serial,\n" +
                "    ( SELECT pj_1.pay_channel\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_channel,\n" +
                "    ( SELECT pj_1.pay_detail\n" +
                "           FROM sdb_pay_info pj_1\n" +
                "          WHERE ((oj.order_serial) = (pj_1.order_serial))\n" +
                "          ORDER BY pj_1.pay_begin_time DESC\n" +
                "         LIMIT 1) AS pay_detail,\n" +
                "    oj.customer_order_no,\n" +
                "    oj.access_mid,\n" +
                "    oj.buss_type,\n" +
                "    oj.market_product_no,\n" +
                "    oj.receive_user_no,\n" +
                "        CASE\n" +
                "            WHEN (oj.trans_kind = '0006') THEN oj.receive_user_no\n" +
                "            ELSE oj.user_no\n" +
                "        END AS user_no,\n" +
                "        CASE\n" +
                "            WHEN (oj.trans_kind = '0006') THEN usr.user_name\n" +
                "            ELSE us.user_name\n" +
                "        END AS user_name,\n" +
                "    usr.user_name AS receive_user_name,\n" +
                "    '' AS trans_rival_type,\n" +
                "    '' AS trans_name,\n" +
                "    '' AS trans_rival_type_account,\n" +
                "    '' AS acc_abstract,\n" +
                "    '' AS acc_product_no,\n" +
                "    oj.merch_order_no,\n" +
                "    '' AS create_time,\n" +
                "    0-oj.order_amt AS acc_amt,\n" +
                "    '' AS postscript,\n" +
                "        CASE oj.trans_kind\n" +
                "            WHEN '0006' THEN oj.orig_merch_order_no\n" +
                "            ELSE oj.ori_customer_order_no\n" +
                "        END AS ori_customer_order_no,\n" +
                "    oj.receive_card_no\n" +
                "   FROM oj2 as oj\n" +
                "     LEFT JOIN ums_info_user_base us ON oj.user_no = us.user_id\n" +
                "     LEFT JOIN ums_info_user_base usr ON oj.receive_user_no = usr.user_id");

        tEnv.executeSql("CREATE TABLE pssdb_reconli_bat_flink (\n" +
                "  id string,\n"+
                "  sand_tn string,\n" +
                "  acc_type string,\n" +
                "  memo string,\n" +
                "  currency string,\n" +
                "  order_title string,\n" +
                "  order_time_out string,\n" +
                "  merch_name string,\n" +
                "  order_end string,\n" +
                "  order_amt DECIMAL(14,2),\n" +
                "  trans_type_in string,\n" +
                "  trans_type_out string,\n" +
                "  order_begin string,\n" +
                "  order_serial string,\n" +
                "  order_desc string,\n" +
                "  trade_status string,\n" +
                "  resp_code string,\n" +
                "  resp_msg string,\n" +
                "  trans_kind string,\n" +
                "  trans_type string,\n" +
                "  front_trans_code string,\n" +
                "  access_system_no string,\n" +
                "  credit_flag string,\n" +
                "  bindid string,\n" +
                "  fee_amt DECIMAL(14,2),\n" +
                "  status string,\n" +
                "  channel_resp_code string,\n" +
                "  channel_resp_msg string,\n" +
                "  pay_amt DECIMAL(14,2),\n" +
                "  rebate_amt DECIMAL(14,2),\n" +
                "  bank_code string,\n" +
                "  channel_order_no string,\n" +
                "  pay_begin_time string,\n" +
                "  pay_tool string,\n" +
                "  pay_end_time string,\n" +
                "  pay_card_no string,\n" +
                "  pay_serial string,\n" +
                "  pay_channel string,\n" +
                "  pay_detail string,\n" +
                "  customer_order_no string,\n" +
                "  access_mid string,\n" +
                "  buss_type string,\n" +
                "  market_product_no string,\n" +
                "  receive_user_no string,\n" +
                "  user_no string,\n" +
                "  user_name string,\n" +
                "  receive_user_name string,\n" +
                "  trans_rival_type string,\n" +
                "  trans_name string,\n" +
                "  trans_rival_type_account string,\n" +
                "  acc_abstract string,\n" +
                "  acc_product_no string,\n" +
                "  merch_order_no string,\n" +
                "  create_time string,\n" +
                "  acc_amt DECIMAL(14,2),\n" +
                "  postscript string,\n" +
                "  ori_customer_order_no string,\n" +
                "  receive_card_no string,\n" +
                "  PRIMARY KEY(id) NOT ENFORCED"+
                ") with("
                + "'connector' = 'jdbc',"
                + "'url' = '"+ CommonConfig.POSTPRE_URL+"',"
                + "'table-name' = 'pssdb_reconli_bat_flinksql',"
                + "'driver' = 'org.postgresql.Driver',"
                + "'username' = '"+ CommonConfig.POSTPRE_USER+"',"
                + "'password' = '"+ CommonConfig.POSTPRE_PWD+"',"
                + "  'sink.buffer-flush.max-rows' = '0',"
                + "  'sink.buffer-flush.interval' = '300ms',"
                + "'sink.max-retries' = '3'"
                + ")");
        //        final Table result =
        // print the result to the console
        tEnv.executeSql("insert into pssdb_reconli_bat_flink select CONCAT_WS('-',order_serial,pay_serial),sand_tn ,acc_type ,memo ,currency ,order_title ,order_time_out ,merch_name ,order_end ,order_amt ,trans_type_in ,trans_type_out ,order_begin ,order_serial ,order_desc ,trade_status ,resp_code ,resp_msg ,trans_kind ,trans_type ,front_trans_code ,access_system_no ,credit_flag ,bindid ,fee_amt ,status ,channel_resp_code ,channel_resp_msg ,pay_amt ,rebate_amt ,bank_code ,channel_order_no ,pay_begin_time ,pay_tool ,pay_end_time ,pay_card_no ,pay_serial ,pay_channel ,pay_detail ,customer_order_no ,access_mid ,buss_type ,market_product_no ,receive_user_no ,user_no ,user_name ,receive_user_name ,trans_rival_type ,trans_name ,trans_rival_type_account ,acc_abstract ,acc_product_no ,merch_order_no ,create_time ,acc_amt ,postscript ,ori_customer_order_no ,receive_card_no  from " +
                "(select * from v_pssdb_reconli \n" +
                "union all\n" +
                "select * from v_pssdb_reconli2 )t").await();
    }
}
